package com.shujia.mr.kqzl1;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapTask;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class KqzkDemo1 {
    public static void main(String[] args) {
        try {
            //创建hadoop配置文件对象
            Configuration conf = new Configuration();
//            conf.set("mapreduce.job.inputformat.class","/xxx/xxx/xx");
//            conf.set("mapreduce.input.fileinputformat.split.minsize","256");
//            conf.set("mapreduce.input.fileinputformat.split.maxsize","200");
            //设置环形缓冲区溢写比例
//            conf.set("mapreduce.map.sort.spill.percent","0.6");
            //修改环形缓冲区大小
//            conf.set("mapreduce.task.io.sort.mb","200");
            //修改reduce个数
//            conf.set("mapreduce.job.reduces","2");
            //键值对数据分区编号的计算逻辑：
            //(key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;


            //创建MapReduce作业对象
            Job job = Job.getInstance(conf);
            //设置mr作业的名字
            job.setJobName("33期空气质量案例1");
            //设置reduce的个数
//            job.setNumReduceTasks(2);

            //设置mr作业的主类
            job.setJarByClass(KqzkDemo1.class);

            //设置map类
            job.setMapperClass(KqzlMapper.class);
            //设置reduce类
            job.setReducerClass(KqzlReducer.class);

            //设置map任务的输出数据类型
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(LongWritable.class);

            //设置reduce任务的输出类型
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(LongWritable.class);

            // 设置读取hdfs文件的路径
//            FileInputFormat.addInputPath(); //添加某一个目录
//            FileInputFormat.setInputPaths(); // 可以添加多个目录
//            FileInputFormat.setInputPaths(job, new Path("/xxx/xxx/xx"));
            FileInputFormat.setInputPaths(job, new Path(args[0]));

            //设置输出的hdfs目录路径
            FileOutputFormat.setOutputPath(job,new Path(args[1]));

            //提交作业到yarn中运行
            boolean b = job.waitForCompletion(true);

            if(b){
                System.out.println("========33期空气质量案例1执行成功！！！！========");
                System.out.println("输出结果目标在hdfs路径："+args[1]);
            }else {
                System.out.println("-------------- 执行失败！！ -----------------");
            }

//            MapTask


        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
